//package com.sunye.kafka;
//
//import com.alibaba.fastjson.JSON;
//import com.fasterxml.jackson.databind.util.JSONPObject;
//import org.apache.kafka.clients.producer.Callback;
//import org.apache.kafka.clients.producer.RecordMetadata;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.kafka.core.KafkaOperations;
//import org.springframework.kafka.core.KafkaTemplate;
//import org.springframework.kafka.retrytopic.DestinationTopic;
//import org.springframework.kafka.support.SendResult;
//import org.springframework.stereotype.Component;
//import org.springframework.transaction.annotation.Transactional;
//import org.springframework.util.concurrent.ListenableFuture;
//import org.springframework.util.concurrent.ListenableFutureCallback;
//import org.springframework.web.bind.annotation.GetMapping;
//import org.springframework.web.bind.annotation.PathVariable;
//import org.springframework.web.bind.annotation.RequestMapping;
//import org.springframework.web.bind.annotation.RestController;
//
//import java.util.*;
//
//@RestController
//@RequestMapping("kafkaController")
//public class KafkaController {
//    @Autowired
//    private KafkaTemplate<String,String> kafkaTemplate;
//
//
//    /**
//     * 生产者发送消息到Batch缓冲区，在一次性发送到broker上去，这样能提高性能。
//     * kafkaTemplate.send的参数代表
//     * 生产者记录（简称PR）的发送逻辑:(Partition分区)
//     * <1> 若指定Partition ID,则PR被发送至指定Partition
//     * <2> 若未指定Partition ID,但指定了Key, PR会按照hasy(key)发送至对应Partition
//     * <3> 若既未指定Partition ID也没指定Key，PR会按照round-robin模式发送到每个Partition
//     * <4> 若同时指定了Partition ID和Key, PR只会发送到指定的Partition (Key不起作用，代码逻辑决定)
//     */
//    @GetMapping("/sendMessage1")
//    @Transactional(rollbackFor = RuntimeException.class,transactionManager = "kafkaTransactionManager")
//    public String senMessage1(){
//        Map map = new HashMap<>();
//        List<String> list = Arrays.asList("111", "222", "333");
//        map.put("list",list);
//        map.put("msg","这是一个消息");
//        String jsonString = JSON.toJSONString(map);
//        kafkaTemplate.send("topic1",jsonString);
//        return "发送成功";
//    }
//
//
//    //携带回调函数
//    @GetMapping("/sendWithCallBackAndPartition")
//    @Transactional(rollbackFor = RuntimeException.class,transactionManager = "kafkaTransactionManager")
//    public String sendWithCallBackAndPartition(){
//        try {
//            for (int i = 0; i < 4; i++) {
//                kafkaTemplate.send("topicWithCall",3,"","拉拉德玛西亚"+i).addCallback(
//                        success ->{
//                            String topic = success.getRecordMetadata().topic();
//                            int partition = success.getRecordMetadata().partition();
//                            long offset = success.getRecordMetadata().offset();
//                            System.out.println("-----------------topic:" + topic + " partition:" + partition + " offset:" + offset);
//                        },
//                        failure ->{
//                            String message1 = failure.getMessage();
//                            System.out.println("错误--------------------"+message1);
//                        }
//                );
//            }
//        }catch (Exception e){
//            e.printStackTrace();
//        }
//        return "发送成功";
//    }
//
//
//
//    //注解事务
//    @GetMapping("/sendTransMessage1")
//    @Transactional(rollbackFor = RuntimeException.class,transactionManager = "kafkaTransactionManager")
//    public String sendTransMessage1(int num){
//        kafkaTemplate.send("topicTrans","这是一个消息");
//        if(num==0){
//            throw new RuntimeException();
//        }
//        kafkaTemplate.send("topicTrans","这是一个消息2");
//        return "发送成功";
//    }
//
//
//    //声明式事务
//    @GetMapping("/sendTransMessage2")
//    public String sendTransMessage2(int num){
//        kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback() {
//            @Override
//            public Object doInOperations(KafkaOperations operations) {
//                kafkaTemplate.send("topicTrans","这是一个消息");
//                if(num==0){
//                    throw new RuntimeException();
//                }
//                kafkaTemplate.send("topicTrans","这是一个消息2");
//                return true;
//            }
//        });
//        return "发送成功";
//    }
//
//
//}
